package com.zlsf.publish;


import com.zlsf.entity.Media;
import com.zlsf.service.SessionManager;
import io.vertx.ext.web.RoutingContext;
import lombok.extern.slf4j.Slf4j;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

@Slf4j
public final class PublishManager {
  ConcurrentHashMap<String, Channel> channels;
  private final ScheduledExecutorService scheduler;

  private PublishManager() {
    channels = new ConcurrentHashMap<String, Channel>();
    scheduler = Executors.newScheduledThreadPool(1);
    scheduler.scheduleAtFixedRate(this::checkChannels, 30, 30, TimeUnit.SECONDS);
  }

  private void checkChannels() {
    for (String tag : channels.keySet()) {
      Channel channel = channels.get(tag);
      if (channel != null) {
        int subscriberCount = channel.getSubscriberCount();
        if (subscriberCount == 0) {
          channel.incrementZeroCount();
          if (channel.getZeroCount() >= 3) {
            SessionManager.removeChannel(tag);
          }
        } else {
          channel.resetZeroCount();
        }
      }
    }
  }

  public Subscriber subscribe(String tag, Media.Type type, RoutingContext ctx) {
    Channel chl = channels.get(tag);
    if (chl == null) {
      chl = new Channel(tag);
      channels.put(tag, chl);
    }
    Subscriber subscriber = null;
    if (type.equals(Media.Type.Video)) subscriber = chl.subscribe(ctx);
    else throw new RuntimeException("unknown media type: " + type);

    subscriber.setName("subscriber-" + tag + "-" + subscriber.getId());
    subscriber.start();

    return subscriber;
  }

  public void publishAudio(String tag, int sequence, long timestamp, int payloadType, byte[] data) {
    Channel chl = channels.get(tag);
    if (chl != null) chl.writeAudio(timestamp, payloadType, data);
  }

  public void publishVideo(String tag, int sequence, long timestamp, int payloadType, byte[] data) {
    Channel chl = channels.get(tag);
    if (chl != null) chl.writeVideo(sequence, timestamp, payloadType, data);
  }

  public Channel open(String tag) {
    Channel chl = channels.get(tag);
    if (chl == null) {
      chl = new Channel(tag);
      channels.put(tag, chl);
    }
    if (chl.isPublishing()) {
      log.info("channel:{}已发布", tag);
    }
    return chl;
  }

  public void close(String tag) {
    Channel chl = channels.remove(tag);
    if (chl != null) chl.close();
  }

  public boolean unsubscribe(String tag, RoutingContext ctx) {
    Channel chl = channels.get(tag);
    if (chl != null) {
      chl.unsubscribe(ctx);
      log.info("web订阅关闭: {} - {}", tag, chl.subscribers.size());
      return chl.subscribers.isEmpty();
    }
    log.info("web订阅关闭: {} - {}", tag, chl.subscribers.size());
    return true;
  }

  static final PublishManager instance = new PublishManager();

  public static void init() {
  }

  public static PublishManager getInstance() {
    return instance;
  }
}
